package com.gxitsky.test;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class MsgConsumer3 {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("admin");
        factory.setPassword("123456");
        factory.setHost("192.168.220.139");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 创建连接
        Connection connection = factory.newConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        channel.basicQos(64);
        // 声明交换器
        String exchangeName = "exchange";
        channel.exchangeDeclare(exchangeName, "direct", true, false, null);

        // 声明一个持久化,非排他,非自动删除的队列
        String queueName = "my-queue";
        channel.queueDeclare(queueName, true, false, false, null);

        String routingKey = "my-queue";
        channel.queueBind(queueName, exchangeName, routingKey);

        while (true) {
            boolean autoAck = false;
//            channel.basicQos(64);
            channel.basicConsume(queueName, autoAck, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey1 = envelope.getRoutingKey();
                    String contentType = properties.getContentType();
                    System.out.println("routing key:" + routingKey1);
                    System.out.println("content type:" + contentType);
                    long deliveryTag = envelope.getDeliveryTag();
                    String msg = new String(body, "UTF-8");
                    System.out.println("消息内容:" + msg);
                    channel.basicAck(deliveryTag, false);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });

        }
    }
}
